#include <cassert>
#include <ctime>
#include <iostream>

#ifndef WIN32
#include <signal.h>
#endif

#include "miniutil/os_compatible.h"
#include "miniutil/socket.h"
#include "miniutil/socketcmn.h"
#include "miniutil/buffer_alloc.h"
#include "miniutil/threadpool.h"
#include "miniutil/socketmultiplex.h"

/*
	SocketMulplexService accept socket instance and try to check if it can read/write/exception
	it will not create socket, close socket, and so on
	Invoker should control all these resources. 

	must call reset_socket_event to continue;
*/
namespace miniutil { namespace multinet {

	MonSocketInstance::MonSocketInstance()
	{
		socevent = EVT_NONE;
		soc = NULL;
		closeflag = false;
		waitevt = EVT_NONE;
	}

	MonTcpSocketInstance::MonTcpSocketInstance()
		:recvbuf(32768*4)
	{
		//psendbuf = NULL;
		//sendlen = 0;
		//lenhavesent = 0;

		recvret = 1;
		sendret = 1;
	}

	void MonTcpSocketInstance::sendbuffer(char* pbuf, int len)
	{
		if ( send_queue.size() > 0 )
		{
			miniutil::data::buffer * last_send_buffer = send_queue.back().get();
			if ( last_send_buffer->free_length() >= (unsigned int)len )
			{
				last_send_buffer->append( pbuf, len );
				return;
			}
		}

		miniutil::data::shared_buffer sbuf = miniutil::data::shared_buffer::safenew( len );
		memcpy( sbuf->data(), pbuf, len );
		sbuf->set_length( len );
		this->send_queue.push_back( sbuf );
	}

	MonUdpSocketInstance::UDPMSG::UDPMSG()
	{
		mlen = 0;
		mpbuf = miniutil::data::pool_buffer_alloc::getdefaultinstance()->retrieve_mem(UDPMSG_SIZE);
		soc = 0;
	}

	MonUdpSocketInstance::UDPMSG::~UDPMSG()
	{
		if(mpbuf != NULL)
		{
			miniutil::data::pool_buffer_alloc::getdefaultinstance()->release_mem(mpbuf);
		}
	}

	//---------------------------------------------------------------

	SocketMulplexService::SocketMulplexService()
	{
#ifdef WIN32
	    WSADATA wsdata;
	    if (WSAStartup(MAKEWORD(1, 1), &wsdata)) return ;
	    //if (WSAStartup(MAKEWORD(2, 0), &wsdata)) return ;
#else
        signal(SIGPIPE,SIG_IGN);
#endif
        stopcmd = false;
		stopstus = true;

		max_dquota = 0x7fffffff;
		max_uquota = 0x7fffffff;

        M_DEBUG("Max QUota:%d\n", max_uquota);
        mprechktime = 0;
	}

	void SocketMulplexService::start_service()
	{
		if(!stopstus)
			return;
		bool bret = miniutil::threadpool::RunTask( this );
		stopstus = false;
	}


	void SocketMulplexService::stop_service()
	{
		if(stopstus)
			return;

		stopcmd = true;
		while(!stopstus)
		{
			miniutil::thread::sleep(10);
		}
	}

	int SocketMulplexService::add_tcp_socket(SOCKET soc, int checkevt)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			malltcpsockets[soc].waitevt = checkevt;
			return 0;
		}
		malltcpsockets[soc] = MonTcpSocketInstance();
		malltcpsockets[soc].soc = soc;
		malltcpsockets[soc].waitevt = checkevt;
		return 0;
	}

	int SocketMulplexService::add_udp_socket(SOCKET soc, int checkevt)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) != malludpsockets.end())
		{
			malludpsockets[soc].waitevt = checkevt;
			return 0;
		}
		malludpsockets[soc] = MonUdpSocketInstance();
		malludpsockets[soc].soc = soc;
		malludpsockets[soc].waitevt = checkevt;
		return 0;
	}

	int SocketMulplexService::del_socket(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			malltcpsockets[soc].closeflag =true;
			return 0;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			malludpsockets[soc].closeflag =true;
			return 0;
		}

		return -1;
	}
	


	//-----------------------------------
	bool SocketMulplexService::is_readable(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			MonTcpSocketInstance& ains = malltcpsockets[soc];
			//if(malltcpsockets[soc].socevent& EVT_READ_READY)
			if(ains.recvbuf.have_content() || ains.recvret <= 0)
				return true;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			MonUdpSocketInstance& ains = malludpsockets[soc];
			//if(malludpsockets[soc].socevent& EVT_READ_READY)
			if(ains.vrecvbuf.size() > 0 )
				return true;
		}
		return false;
	}

	bool SocketMulplexService::is_writeable(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) != malltcpsockets.end())
		{
			//if(malltcpsockets[soc].socevent& EVT_WRITE_READY)
			//if(malltcpsockets[soc].psendbuf == NULL)
				return true;
		}
		else if(malludpsockets.find(soc) != malludpsockets.end())
		{
			//if(malludpsockets[soc].socevent& EVT_WRITE_READY)
				return true;
		}
		return false;
	}

	bool SocketMulplexService::is_incoming(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_INCOMING)
        {
            //before accept, keep this flag
            //malltcpsockets[soc].socevent &= (~EVT_INCOMING);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_exception(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_EXCEPTION)
        {
            malltcpsockets[soc].waitevt &= (~EVT_EXCEPTION);
            malltcpsockets[soc].socevent &= (~EVT_EXCEPTION);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_connected(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		if(malltcpsockets[soc].socevent& EVT_CONNECTED)
        {
            malltcpsockets[soc].waitevt &= (~EVT_CONNECTED);
            malltcpsockets[soc].socevent &= (~EVT_CONNECTED);
			return true;
        }
		return false;
	}

	bool SocketMulplexService::is_closed(SOCKET soc)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return false;
		//if(malltcpsockets[soc].socevent& EVT_CLOSED)
		if(malltcpsockets[soc].recvret <= 0)
			return true;
		return false;
	}

    int SocketMulplexService::check_tcp_connect(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_CONNECTED | EVT_EXCEPTION);
    }

    int SocketMulplexService::check_tcp_listen(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_INCOMING);
    }

    int SocketMulplexService::check_tcp_read(SOCKET soc)
    {
        return add_tcp_socket(soc, EVT_READ_READY);
    }

	//----------------------------------------------------------------------------

	//return  -2 other error
	// -1 socket error
	// > 0 ok
	//notice, must have event EVT_WRITE_READY
	int SocketMulplexService::send(SOCKET soc, const char* buf, int len)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -2;
		MonTcpSocketInstance &ins = malltcpsockets[soc];
		//if(ins.psendbuf != NULL)
		//	return -2;
		ins.sendbuffer(const_cast<char*>(buf), len);

		////assert(ins.socevent & EVT_WRITE_READY);

		//if(ins.sendret <= 0)  //last time send error, so reject to send again.
		//	return ins.sendret;
		//ins.psendbuf = buf;
		//ins.sendlen = len;
		//ins.lenhavesent = 0;

		////ins.socevent &= (~EVT_WRITE_READY);
		return len;
	}

	//return <=0 socket error
	//notice, must have event EVT_READ_READY
	int SocketMulplexService::recv(SOCKET soc, char* buf, int len)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -1;

		MonTcpSocketInstance &ins = malltcpsockets[soc];

		if(ins.recvbuf.have_content())
			return ins.recvbuf.read(buf, len);

		if(ins.recvret <= 0)
			return ins.recvret;

        M_DEBUG("%d",ins.recvret)
		assert(0);
		//assert(ins.socevent & EVT_READ_READY);
		//if(ins.lenhavereceived < ins.recvlen)
		//{
		//	int surlen = ins.recvlen - ins.lenhavereceived;
		//	int cplen = surlen>len?len:surlen;
		//	memcpy(buf, ins.recvbuf+ins.lenhavereceived, cplen);
		//	ins.lenhavereceived += cplen;
		//	if(ins.lenhavereceived == ins.recvlen)
		//	{
		//		ins.lenhavereceived = 0;
		//		ins.recvlen = 0;
		//		ins.socevent &= (~EVT_READ_READY);
		//	}
		//	return cplen;
		//}

		////assert(0);
		//return -2;
		return 0;
	}

	SOCKET SocketMulplexService::accept(SOCKET soc, sockaddr* addr, int *addrlen)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malltcpsockets.find(soc) == malltcpsockets.end())
			return -1;

		MonSocketInstance &ins = malltcpsockets[soc];
#ifdef WIN32
		SOCKET connector = ::accept(soc, addr , addrlen );
#else
		SOCKET connector = ::accept(soc, addr , (socklen_t*)addrlen );
#endif
		ins.socevent &= (~EVT_INCOMING);
		return connector;
	}

	int SocketMulplexService::sendto(SOCKET soc, const char* buf,int len,const struct sockaddr* to)
	{
        if(len > MonUdpSocketInstance::UDPMSG_SIZE)
			return -1;
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) == malludpsockets.end())
			return -1;

		MonUdpSocketInstance::UDPMSG* pmsg = new MonUdpSocketInstance::UDPMSG();
		memcpy(pmsg->mpbuf, buf, len);
		pmsg->mlen = len;
		pmsg->maddr = *to;
		pmsg->soc = soc;
		malludpsockets[soc].vsendbuf.push_back(pmsg);
		return len;
	}

	int SocketMulplexService::recvfrom(SOCKET soc, char *buf, int len, sockaddr *from)
	{
		miniutil::auto_lock kk(&msockmtx);
		if(malludpsockets.find(soc) == malludpsockets.end())
			return -1;

		if(malludpsockets[soc].vrecvbuf.size() == 0)
			return 0;

		int finallen = len;
		if(len < malludpsockets[soc].vrecvbuf[0]->mlen)
			finallen = malludpsockets[soc].vrecvbuf[0]->mlen;

		memcpy(buf, malludpsockets[soc].vrecvbuf[0]->mpbuf, finallen);

		//drop excess data
		delete malludpsockets[soc].vrecvbuf[0];
		malludpsockets[soc].vrecvbuf.erase(malludpsockets[soc].vrecvbuf.begin());

		return finallen;
	}

	int SocketMulplexService::limitnetspeed(int quota, int direction)	//byte quota and directory
	{
		if(direction == 0)
			max_dquota = quota;
		else
			max_uquota = quota;
        M_DEBUG("limitnetspeed Max QUota:%d\n", max_uquota);
		return 0;
	}

	//---------------------------------------------------------------
	void SocketMulplexService_Select::run()
	{
	
		timeval tv;
		tv.tv_sec = 0;
		tv.tv_usec = 1000*10;
		
		stopstus =  false;

		while(true)
		{
			if(stopcmd)
			{
				stopcmd = false;
				break;
			}

			do
			{
				thread::sleep( 10 );
				miniutil::auto_lock kk(&msockmtx);

				SOCKET max_socket = init_fdset();
				
				int ret = select((int)(max_socket+1), &readdfs, &writedfs, &excepdfs, &tv);
				if ( ret < 0 )
				{
		#ifdef WIN32
					int sret = WSAGetLastError();
		#endif
					continue;
				}
				else if ( ret == 0 ) 
				{
					continue;
				}

				time_t now;
				time(&now);
				if(now > mprechktime)
				{
                    //M_DEBUG("Restore QUota:%d\n", max_uquota);
					left_dquota = max_dquota;
					left_uquota = max_uquota;
					mprechktime = now;
				}
				
				if(left_dquota > 0)
					this->do_readable( );
				if(left_uquota > 0)
					this->do_writable( );

				do_exception();
			}while(0);

			miniutil::thread::sleep(1);
		}
		stopstus = true;
	}

	SOCKET SocketMulplexService_Select::init_fdset()
	{
		SOCKET max_socket = 0;
		FD_ZERO(&writedfs);
		FD_ZERO(&readdfs);
		FD_ZERO(&excepdfs);	
		

		tcpchecksocs.clear();
		udpchecksocs.clear();
		do  //add all socket to ...
		{
			miniutil::auto_lock kk(&msockmtx);
			std::map<SOCKET, MonTcpSocketInstance>::iterator iter_vsmt;
			for(iter_vsmt = malltcpsockets.begin(); iter_vsmt != malltcpsockets.end();)
			{
				if(iter_vsmt->second.closeflag)
				{
					malltcpsockets.erase(iter_vsmt++);
					continue;
				}
				MonTcpSocketInstance &ains = iter_vsmt->second;

				//if(iter_vsm->second.socevent != EVT_NONE)
				//{
				//	iter_vsm ++;
				//	continue;
				//}

				tcpchecksocs.push_back(iter_vsmt->first);
				if((ains.waitevt & EVT_INCOMING ) != 0x00 &&
					(ains.socevent & EVT_INCOMING) == 0x00)
					FD_SET(iter_vsmt->first, &readdfs);

				if((ains.waitevt & EVT_READ_READY) != 0x00 &&
					ains.recvret > 0 &&
					ains.recvbuf.have_space(1024))
				{
					//if(iter_vsm->second.recvlen > 0) //still data there
					//	iter_vsm->second.socevent |= EVT_READ_READY;
					//else
						FD_SET(iter_vsmt->first, &readdfs);
				}
				if((ains.waitevt & EVT_CONNECTED) != 0x00 &&
					(ains.socevent & EVT_CONNECTED) == 0x00)
                {
					FD_SET(iter_vsmt->first, &writedfs);
                    FD_SET(iter_vsmt->first, &excepdfs);
                }

				if(/*(iter_vsmt->second.waitevt & EVT_WRITE_READY) != 0x00 &&*/
					ains.sendret > 0 &&
					ains.send_queue.size() > 0)
					FD_SET(iter_vsmt->first, &writedfs);

				if((ains.waitevt & EVT_EXCEPTION) != 0x00 &&
					(ains.socevent & EVT_EXCEPTION) == 0x00)
					FD_SET(iter_vsmt->first, &excepdfs);

				if(iter_vsmt->first > max_socket)
					max_socket = iter_vsmt->first;
				iter_vsmt ++;
			}

			std::map<SOCKET, MonUdpSocketInstance>::iterator iter_msmu;
			for(iter_msmu = malludpsockets.begin(); iter_msmu != malludpsockets.end();)
			{
				if(iter_msmu->second.closeflag)
				{
					malludpsockets.erase(iter_msmu++);
					continue;
				}
				//low eff
				if(iter_msmu->second.vsendbuf.size() > 0)
				{
					FD_SET(iter_vsmt->first, &writedfs);
				}
				FD_SET(iter_msmu->first, &readdfs);
				udpchecksocs.push_back(iter_msmu->first);
				iter_msmu ++;
			}

		}while(0);
		return max_socket;
	}

	void SocketMulplexService_Select::do_readable()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], &readdfs))
			{
				if((ins.waitevt & EVT_INCOMING) != 0x00)
					ins.socevent |= EVT_INCOMING;
				else
				{
					std::pair<char*, unsigned int> writepos;
					ins.recvbuf.getwritepos(writepos);
					assert(writepos.second > 0);
					{
						ins.recvret = ::recv(ins.soc, writepos.first,writepos.second, 0);
						if(ins.recvret > 0)
						{
							ins.recvbuf.more_content(ins.recvret);
							left_dquota -= ins.recvret;
							if(left_dquota < 0)
								break;
						}
						else
						{
							int ret = miniutil::socket::getLastError();
                            M_DEBUG("recvret < 0 ret:%d, err:%s\n", ret, miniutil::socketcmn::get_soc_errorstr(ret).c_str());
						}
					}
					ins.socevent |= EVT_READ_READY;
				}
			}
		}
		for(unsigned int i=0; i< udpchecksocs.size(); i++)
		{
			socklen_t nmln = sizeof(struct sockaddr);
			MonUdpSocketInstance &ins = malludpsockets[udpchecksocs[i]];
			if(FD_ISSET(udpchecksocs[i], &readdfs))
			{
				MonUdpSocketInstance::UDPMSG* pmsg = new MonUdpSocketInstance::UDPMSG();
				pmsg->mlen = ::recvfrom(udpchecksocs[i], pmsg->mpbuf, MonUdpSocketInstance::UDPMSG_SIZE, 
                    0, (struct sockaddr*)&pmsg->maddr, &nmln);
				ins.vrecvbuf.push_back(pmsg);
			}
		}
	}

	void SocketMulplexService_Select::do_writable()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], &writedfs))
			{
				if((ins.waitevt & EVT_CONNECTED) != 0x00)
					ins.socevent |= EVT_CONNECTED;
				else if(/*(ins.waitevt & EVT_WRITE_READY) != 0x00*/ ins.send_queue.size() > 0)
				{
					while ( ins.send_queue.size() > 0 )
					{
                        //M_DEBUG("ins.send_queue.size():%d\n", ins.send_queue.size());
						miniutil::data::shared_buffer b = ins.send_queue.front();
						int sendlen =  (int)b->length() > left_uquota?left_uquota:(int)b->length();
						if(sendlen <= 0)
							break;

                        //M_DEBUG("left quota:%d\n", left_uquota);
						int len = ::send( tcpchecksocs[i], b->data(), (int)b->length(), 0 );
						if ( len == (int)b->length() )
						{
							//M_DEBUG("do_send( )len %d == length()%d\n", len, (int)b->length());
							ins.send_queue.pop_front();
							left_uquota -= len;
							if(left_uquota <= 0)
                            {
                                M_DEBUG("No UP quota available:%d %d\n", left_uquota, len);
								break;
                            }
							continue;
						}
						if ( len < 0)
						{
							int ret = miniutil::socket::getLastError();
							if(ret != EWOULDBLOCK)
                            {
                                ins.sendret = -1;
								ins.socevent |= EVT_WRITE_READY;
                                M_DEBUG("send_len < 0 ret:%d, err:%s\n", ret, miniutil::socketcmn::get_soc_errorstr(ret).c_str());
                            }
						}
						else if ( len > 0 && len <  (int)b->length() )
						{
							//M_DEBUG("do_send( )len %d<length()%d\n", len, (int)b->length());
							b->erase( 0 , (size_t)len );
							left_uquota -= len;
						}
						else if(len == 0)
						{
							ins.socevent |= EVT_WRITE_READY;
						}
						break;
					}
				}
			}
		}
		for(unsigned int i=0; i< udpchecksocs.size(); i++)
		{
			socklen_t nmln = sizeof(struct sockaddr);
			MonUdpSocketInstance &ins = malludpsockets[udpchecksocs[i]];
			if(FD_ISSET(udpchecksocs[i], &writedfs))
			{
				if(ins.vrecvbuf.size() > 0)
				{
					MonUdpSocketInstance::UDPMSG* pmsg = ins.vrecvbuf[0];
					::sendto(ins.soc, pmsg->mpbuf, pmsg->mlen, 0, (struct sockaddr*)&pmsg->maddr, nmln);
					delete pmsg;
					ins.vrecvbuf.erase(ins.vrecvbuf.begin());
				}
			}
		}
	}

	void SocketMulplexService_Select::do_exception()
	{
		for(unsigned int i=0; i< tcpchecksocs.size(); i++)
		{
			MonTcpSocketInstance &ins = malltcpsockets[tcpchecksocs[i]];
			if(FD_ISSET(tcpchecksocs[i], &excepdfs))
				ins.socevent |= EVT_EXCEPTION;
		}
	}

} }
